package org.tao.lightningmq.broker.core;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.model.CommitLogModel;
import org.tao.lightningmq.broker.model.ConsumeQueueDetailModel;
import org.tao.lightningmq.broker.model.MqTopicModel;
import org.tao.lightningmq.broker.model.QueueModel;
import org.tao.lightningmq.broker.utils.LogFileNameUtil;
import org.tao.lightningmq.broker.utils.PutMessageLock;
import org.tao.lightningmq.broker.utils.UnfairReentrantLock;
import org.tao.lightningmq.common.constants.BrokerConstants;
import org.tao.lightningmq.common.dto.ConsumeMsgCommitLogDTO;
import org.tao.lightningmq.common.dto.MessageDTO;
import org.tao.lightningmq.common.enums.FileSizeUnitEnum;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author lht
 * @date 2025/1/3 - 16:21
 * @description: mmap对象模型
 */
public class CommitLogMMapFileModel {

    private static final Logger LOGGER = LoggerFactory.getLogger(CommitLogMMapFileModel.class);

    private File file;
    private MappedByteBuffer mappedByteBuffer;
    private ByteBuffer readByteBuffer;
    private String topic;
    private PutMessageLock putMessageLock;

    public void loadFileInMMap(String topic, int startOffset, int mappedSize, FileSizeUnitEnum fileSizeUnitEnum) throws IOException {
        this.topic = topic;
        String filePath = getLatestCommitLogFile(topic);
        doMMap(filePath, startOffset, mappedSize, fileSizeUnitEnum);
        putMessageLock = new UnfairReentrantLock();
    }

    private void doMMap(String filePath, int startOffset, int mappedSize, FileSizeUnitEnum fileSizeUnitEnum) throws IOException {
        FileSizeUnitEnum fileSize = fileSizeUnitEnum == null ? FileSizeUnitEnum.B : fileSizeUnitEnum;
        file = new File(filePath);
        if (!file.exists()) {
            throw new RuntimeException("filePath is " + filePath + " inValid");
        }
        mappedSize *= fileSize.getSize();
        FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
        mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, startOffset, mappedSize);
        this.readByteBuffer = mappedByteBuffer.slice();
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        this.mappedByteBuffer.position(mqTopicModel.getCommitLogModel().getOffset().get());
    }

    private String getLatestCommitLogFile(String topic) {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        if (mqTopicModel == null) {
            throw new RuntimeException("topic is " + topic + " inValid");
        }
        CommitLogModel commitLogModel = mqTopicModel.getCommitLogModel();
        Long diff = commitLogModel.countDiff();
        String filePath = null;
        if (diff > 0) {
            filePath = LogFileNameUtil.buildCommitLogFilePath(topic, commitLogModel.getFileName());
        } else {
            CommitLogFilePath commitLogFilePath = createNewCommitLogFile(topic, commitLogModel);
            filePath = commitLogFilePath.getFilePath();
        }
        return filePath;
    }

    public void writeContent(MessageDTO messageDTO) throws IOException {
        writeContent(messageDTO, false);
    }

    public void writeContent(MessageDTO messageDTO, boolean force) throws IOException {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        if (mqTopicModel == null) {
            throw new IllegalArgumentException("eagleMqTopicModel is null");
        }
        CommitLogModel commitLogModel = mqTopicModel.getCommitLogModel();
        if (commitLogModel == null) {
            throw new IllegalArgumentException("commitLogModel is null");
        }
        putMessageLock.lock();
        try {
            checkCommitLogHasEnableSpace(messageDTO.getBody(), mqTopicModel);
            this.mappedByteBuffer.put(messageDTO.getBody());
            AtomicInteger offset = commitLogModel.getOffset();
            this.dispatcher(messageDTO, offset.get());
            offset.addAndGet(messageDTO.getBody().length);
            if (force) {
                this.mappedByteBuffer.force();
            }
        } finally {
            putMessageLock.unlock();
        }
    }

    private void dispatcher(MessageDTO messageDTO, int msgIdx) {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        if (mqTopicModel == null) {
            throw new RuntimeException("topic is undefined");
        }
        int queueId = 0;
        int queueSize = mqTopicModel.getQueueList().size();
        if (messageDTO.getQueueId() < 0) {
            queueId = new Random().nextInt(queueSize);
        } else if (messageDTO.getQueueId() < queueSize) {
            queueId = messageDTO.getQueueId();
        }
        // 重新设置到messageDTO中，保证同步给从节点时写入的是同一条队列
        messageDTO.setQueueId(queueId);

        ConsumeQueueDetailModel consumeQueueDetailModel = new ConsumeQueueDetailModel();
        consumeQueueDetailModel.setCommitLogFilename(Integer.parseInt(mqTopicModel.getCommitLogModel().getFileName()));
        consumeQueueDetailModel.setMsgIndex(msgIdx);
        consumeQueueDetailModel.setMsgLen(messageDTO.getBody().length);
        consumeQueueDetailModel.setRetryTimes(messageDTO.getCurrentRetryTimes());
        byte[] queueContent = consumeQueueDetailModel.convertToBytes();
        LOGGER.info("[CommitLogMMapFileModel]类,{dispatcher}方法,写入数据:{}", JSON.toJSONString(consumeQueueDetailModel));
        List<ConsumeQueueMMapFileModel> queueMMapFileModels = CommonCache.getConsumeQueueMMapFileModelManager().get(topic);
        ConsumeQueueMMapFileModel consumeQueueMMapFileModel = queueMMapFileModels.get(queueId);
        consumeQueueMMapFileModel.writeContent(queueContent);
        QueueModel queueModel = mqTopicModel.getQueueList().get(queueId);
        queueModel.getLatestOffset().addAndGet(queueContent.length);
    }


    //public byte[] readContent(int readOffset, int size) {
    //    ByteBuffer slice = readByteBuffer.slice();
    //    slice.position(readOffset);
    //    byte[] content = new byte[size];
    //    readByteBuffer.get(content);
    //    return content;
    //}
    /**
     * 从文件的指定offset开始读取内容
     * @param readOffset 读取的起始offset
     * @param size 希望读取的长度
     * @return 读取到的内容
     */
    public ConsumeMsgCommitLogDTO readContent(int readOffset, int size) {
        ByteBuffer readBuf = readByteBuffer.slice();
        readBuf.position(readOffset);
        byte[] readBytes = new byte[size];
        readBuf.get(readBytes);
        ConsumeMsgCommitLogDTO consumeMsgCommitLogDTO = new ConsumeMsgCommitLogDTO();
        consumeMsgCommitLogDTO.setBody(readBytes);
        consumeMsgCommitLogDTO.setFileName(file.getName());
        consumeMsgCommitLogDTO.setCommitLogOffset(readOffset);
        consumeMsgCommitLogDTO.setCommitLogSize(size);
        return consumeMsgCommitLogDTO;
    }

    /**
     * 检查commitlog是否有足够的空间
     * @param content 内容
     * @throws IOException
     */
    private void checkCommitLogHasEnableSpace(byte[] content, MqTopicModel mqTopicModel) throws IOException {
        CommitLogModel commitLogModel = mqTopicModel.getCommitLogModel();
        Long diff = commitLogModel.countDiff();
        if (diff < content.length) {
            LOGGER.info("commitLog剩余容量不足，生成新的文件");
            clean();
            CommitLogFilePath commitLogFilePath = createNewCommitLogFile(topic, commitLogModel);
            commitLogModel.setOffsetLimit(Long.valueOf(BrokerConstants.COMMIT_LOG_DEFAULT_MMAP_SIZE));
            commitLogModel.setOffset(new AtomicInteger(0));
            commitLogModel.setFileName(commitLogFilePath.getFileName());
            this.doMMap(commitLogFilePath.getFilePath(), 0, BrokerConstants.COMMIT_LOG_DEFAULT_MMAP_SIZE, null);
        }
    }

    /**
     * 释放mmap内存占用
     */
    public void clean() {
        if (mappedByteBuffer == null || !mappedByteBuffer.isDirect() || mappedByteBuffer.capacity() == 0)
            return;
        invoke(invoke(viewed(mappedByteBuffer), "cleaner"), "clean");
    }

    private Object invoke(final Object target, final String methodName, final Class<?>... args) {
        return AccessController.doPrivileged(new PrivilegedAction<Object>() {
            public Object run() {
                try {
                    Method method = method(target, methodName, args);
                    method.setAccessible(true);
                    return method.invoke(target);
                } catch (Exception e) {
                    throw new IllegalStateException(e);
                }
            }
        });
    }

    private Method method(Object target, String methodName, Class<?>[] args)
            throws NoSuchMethodException {
        try {
            return target.getClass().getMethod(methodName, args);
        } catch (NoSuchMethodException e) {
            return target.getClass().getDeclaredMethod(methodName, args);
        }
    }

    private ByteBuffer viewed(ByteBuffer buffer) {
        String methodName = "viewedBuffer";
        Method[] methods = buffer.getClass().getMethods();
        for (int i = 0; i < methods.length; i++) {
            if (methods[i].getName().equals("attachment")) {
                methodName = "attachment";
                break;
            }
        }

        ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
        if (viewedBuffer == null)
            return buffer;
        else
            return viewed(viewedBuffer);
    }


    private static class CommitLogFilePath {
        private String fileName;
        private String filePath;

        public CommitLogFilePath(String fileName, String filePath) {
            this.fileName = fileName;
            this.filePath = filePath;
        }

        public String getFileName() {
            return fileName;
        }

        public void setFileName(String fileName) {
            this.fileName = fileName;
        }

        public String getFilePath() {
            return filePath;
        }

        public void setFilePath(String filePath) {
            this.filePath = filePath;
        }

        @Override
        public String toString() {
            return "CommitLogFilePath{" +
                    "fileName='" + fileName + '\'' +
                    ", filePath='" + filePath + '\'' +
                    '}';
        }
    }

    private CommitLogFilePath createNewCommitLogFile(String topic, CommitLogModel commitLogModel) {
        String newFileName = LogFileNameUtil.incrConsumeQueueFileName(commitLogModel.getFileName());
        String newFilePath = LogFileNameUtil.buildCommitLogFilePath(topic, newFileName);
        File newCommitLogFile = new File(newFilePath);
        try {
            boolean newFile = newCommitLogFile.createNewFile();
            if (!newFile) {
                throw new RuntimeException("create new commitlog file failed");
            }
            LOGGER.info("[CommitLogMMapFileModel]类,{createNewCommitLogFile}方法,89行, 创建了新的commitlog文件,文件路径为:{}", newFilePath);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return new CommitLogFilePath(newFileName, newFilePath);
    }

    public File getFile() {
        return file;
    }
}
